package com.min.gps.service.nio;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.min.gps.model.DataCode;
import com.min.gps.model.GPSInfo;
import com.min.gps.model.LoraInfo;
import com.min.gps.service.GPSService;
import com.min.gps.service.LoraService;
import com.min.gps.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

public class SocketServer {

	private static final Logger logger = LoggerFactory.getLogger(SocketServer.class);

	/**
	 * 创建非阻塞服务器绑定2020端口
	 */
	public SocketServer() {
		try {
			ss = ServerSocketChannel.open();
			Integer port = 1122;
			ss.bind(new InetSocketAddress(port));
			ss.configureBlocking(false);
			selector = Selector.open();
			ss.register(selector, SelectionKey.OP_ACCEPT);
		} catch (Exception e) {
			logger.error(e.getMessage());
			close();
		}
	}

	private Map<Integer, Integer> rejectedThreadCountMap = new ConcurrentHashMap<>();

	private RejectedExecutionHandler rejectedExecutionHandler = new RejectedExecutionHandler() {
		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			try {
				TimeUnit.MILLISECONDS.sleep(100);
				int hashcode = r.hashCode();
				Integer count = rejectedThreadCountMap.get(hashcode);
				if (count == null) {
					count = 0;
					rejectedThreadCountMap.put(hashcode, count);
				} else {
					count++;
					rejectedThreadCountMap.put(hashcode, count);
				}
				if (count < 1) {
					executor.execute(r);
				} else {
					if (r instanceof WriteClientSocketHandler) {
						WriteClientSocketHandler realThread = (WriteClientSocketHandler) r;
						logger.info("服务系统繁忙,客户端WriteClientSocketHandler[" + realThread.client + "]请求被拒绝处理！");

						SelectionKey selectionKey = realThread.client.keyFor(selector);
						if (selectionKey != null) {
							selectionKey.cancel();
						}
						if (realThread.client != null) {
							try {
								realThread.client.close();
							} catch (IOException e) {
								e.printStackTrace();
							}
						}
						AtomicBoolean isWriting = isWritingMap.get(realThread.client);
						isWriting.set(false);
					} else if (r instanceof ReadClientSocketHandler) {
						ReadClientSocketHandler realThread = (ReadClientSocketHandler) r;
						logger.info("服务系统繁忙,客户端ReadClientSocketHandler[" + realThread.client + "]请求被拒绝处理！");
						SelectionKey selectionKey = realThread.client.keyFor(selector);
						if (selectionKey != null) {
							selectionKey.cancel();
						}
						if (realThread.client != null) {
							try {
								realThread.client.close();
							} catch (IOException e) {
								e.printStackTrace();
							}
						}
						AtomicBoolean isReading = isReadingMap.get(realThread.client);
						isReading.set(false);
					} else {
						logger.info("服务系统繁忙,系统线程[" + r.getClass().getName() + "]被拒绝处理！");
					}
				}
			} catch (InterruptedException e) {
				logger.info("RejectedExecutionHandler处理发生异常：" + e.getMessage());
			}
		}
	};

	private ThreadPoolExecutor threadPool = new ThreadPoolExecutor(30, 50, 300, TimeUnit.MILLISECONDS,
			new ArrayBlockingQueue<Runnable>(5), rejectedExecutionHandler);
	private Map<SocketChannel, AtomicBoolean> isReadingMap = new ConcurrentHashMap<SocketChannel, AtomicBoolean>();
	private Map<SocketChannel, AtomicBoolean> isWritingMap = new ConcurrentHashMap<SocketChannel, AtomicBoolean>();
	private Selector selector = null;
	private ServerSocketChannel ss = null;
	private volatile boolean isClose = false;

	public boolean isClose() {
		return isClose;
	}

	/**
	 * 关闭服务器
	 */
	private void close() {
		isClose = true;
		threadPool.shutdown();
		try {
			if (ss != null) {
				ss.close();
			}
			if (selector != null) {
				selector.close();
			}
		} catch (IOException e) {
			logger.error("服务器关闭发生异常：" + e.getMessage());
		}
	}

	/**
	 * 启动选择器监听客户端事件
	 */
public void start() {
		threadPool.execute(new SelectorGuardHandler());
	}

	/**
	 * Nio 通道选择器
	 * 
	 * @author cgli
	 *
	 */
	private class SelectorGuardHandler implements Runnable {

		@Override
		public void run() {

			while (!isClose) {
				try {
					if (selector.select(10) == 0) {
						continue;
					}
					Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
					while (iterator.hasNext()) {
						SelectionKey selectedKey = iterator.next();
						iterator.remove();
						try {
							if (selectedKey.isReadable()) {
								SocketChannel socketChannel = (SocketChannel) selectedKey.channel();
								AtomicBoolean isReading = isReadingMap.get(socketChannel);
								if (isReading == null) {
									isReading = new AtomicBoolean(false);
									isReadingMap.put(socketChannel, isReading);
								}
								while (isReading.get()) {
									Thread.sleep(5);
								}
								isReading.set(true);
								threadPool.execute(new ReadClientSocketHandler(socketChannel));

							} else if (selectedKey.isWritable()) {
								Object responseMessage = selectedKey.attachment();
								SocketChannel socketChannel = (SocketChannel) selectedKey.channel();
								selectedKey.interestOps(SelectionKey.OP_READ);
								logger.info("---------------------->isWritable");
								threadPool.execute(new WriteClientSocketHandler(socketChannel, responseMessage));

							} else if (selectedKey.isAcceptable()) {
								ServerSocketChannel ssc = (ServerSocketChannel) selectedKey.channel();
								SocketChannel clientSocket = ssc.accept();
								clientSocket.configureBlocking(false);
								logger.info("---------------------->isAcceptable");
								clientSocket.register(selector, SelectionKey.OP_READ);
							}
						} catch (CancelledKeyException e) {
							selectedKey.cancel();
							logger.error("服务器启动或运行发生异常：" + e);
						}
					}
				} catch (Exception e) {
					if (e instanceof NullPointerException) {
						e.printStackTrace();
						logger.error(e.getMessage());
						close();
					} else {
						logger.error("服务器启动或运行发生异常：" + e);
						close();
					}
					break;
				}
			}
		}
	}

	/**
	 * 
	 * 响应数据给客户端线程
	 * 
	 */
	private class WriteClientSocketHandler implements Runnable {
		private SocketChannel client;

		private Object responseMessage;

		private WriteClientSocketHandler(SocketChannel client, Object responseMessage) {
			this.client = client;
			this.responseMessage = responseMessage;
		}

		@Override
		public void run() {
			try {
				logger.info("------------------w1 准备写入数据-------------");
				byte[] responseByteData = null;
				String logResponseString = "";
				if (responseMessage instanceof byte[]) {
					responseByteData = (byte[]) responseMessage;
					logger.info("---------------w2 写字节数据-------------");
					logResponseString = DataFormater.bytesToHexString(responseByteData);
				} else if (responseMessage instanceof String) {
					logResponseString = (String) responseMessage;
					responseByteData = logResponseString.getBytes();
				} else if (responseMessage != null) {
					logger.info("不支持的数据类型" + responseMessage.getClass());
					return;
				}
				if (responseByteData == null || responseByteData.length == 0) {
					logger.info("服务器响应的写入数据为空");
					return;
				}
				client.write(ByteBuffer.wrap(responseByteData));
                 logger.info("服务端响应客户端[" + client.getRemoteAddress() + "]写入数据 :[" + logResponseString + "]");
			} catch (Exception e) {
				try {
					logger.info("server响应客户端[" + client.getRemoteAddress() + "]数据 异常[" + e.getMessage() + "]");
					SelectionKey selectionKey = client.keyFor(selector);
					if (selectionKey != null) {
						selectionKey.cancel();
					}
					if (client != null) {
						client.close();
					}
				} catch (IOException e1) {
					e1.printStackTrace();
				}
			} finally {
				AtomicBoolean isWriting = isWritingMap.get(client);
				isWriting.set(false);
			}
		}
	}

	/**
	 * 读客户端发送数据线程
	 * 
	 */
	private class ReadClientSocketHandler implements Runnable {

		private SocketChannel client;
		private ByteBuffer dataLen = ByteBuffer.allocate(4);

		private ReadClientSocketHandler(SocketChannel client) {
			this.client = client;
		}

		@Override
		public void run() {
			try {
				dataLen.clear();
				client.read(dataLen);
				if (dataLen.array().length > 0 && dataLen.hasArray()) {

					String slen = DataFormater.byteToHex(dataLen.get(3));
					int dataLength = Integer.parseInt(slen, 16);
					if (dataLength <= 0) {
						return;
					}

					ByteBuffer data = ByteBuffer.allocate(dataLength);//读取到客户端传来的数据
					while (data.hasRemaining()) {
						client.read(data);
					}
					byte[] buff = data.array();
					int type = buff[6];
					byte[] lengthData = new byte[2];//数据长度
					System.arraycopy(buff, 1, lengthData, 0, 2);
					byte[] cityID = new byte[2];//车辆ID
					System.arraycopy(buff, 4, cityID, 0, 2);
					int datalength = Integer.parseInt(ByteUtils.bytesToHexString(lengthData), 16);//字节数组转十进制
					int crclength = datalength + 4;
					byte[] crxData = new byte[crclength];//校验数据
					System.arraycopy(buff, 0, crxData, 0, crclength);
					byte[] Data = new byte[datalength - 3];//解析数据
					System.arraycopy(buff, 7, Data, 0, Data.length);
					byte[] cxcCode = new byte[2];
					System.arraycopy(buff, crclength, cxcCode, 0, 2);
					String checkCode = String.format("%04x", CRC16Util.calcCrc16(crxData)).toUpperCase();
					if (ByteUtils.bytesToHexString(cxcCode).equals(checkCode)) {//crc16校验
						if (type == 1) {//GPS数据解析
							GPSInfo parseData = new GPSInfo();
							String tempdata = new String(Data);
							String gpsArray[] = tempdata.split(",");
							parseData.setCityID(ByteUtils.bytesToHexString(cityID));
							parseData.setSendTime(DateUtil.getNowDateTime());
							parseData.setDataFormat(gpsArray[0]);
							parseData.setStatus(gpsArray[2]);
							parseData.setGPSTime(GpsUtil.posDate(gpsArray[9]) + GpsUtil.posTime(gpsArray[1]));
							parseData.setLatitude(GpsUtil.hemisphere(gpsArray[4]) + GpsUtil.latitude(gpsArray[3]));
							parseData.setLongitude(GpsUtil.hemisphere(gpsArray[6]) + GpsUtil.longitude(gpsArray[5]));
							parseData.setSpeed(gpsArray[7]);
							parseData.setHeading(gpsArray[8]);
							parseData.setMagneticDeclination(gpsArray[10]);
							parseData.setMagneticDirection(gpsArray[11]);
							GPSService dataService = (GPSService) SpringContextUtil.getBean("GPSService");
							dataService.save(parseData);
							//gps数据解析后推送到web端
							JSONObject result = new JSONObject();
							result.put("data", JSON.toJSONString(parseData));
							HttpRequestUtils.httpPost("http://localhost:9001/send", result, false);
							//out.writeUTF(JSON.toJSONString(parseData));
							// out.flush();
						}
						if (type == 2) {//Lora数据解析
							List<DataCode> listCode = new ArrayList<>();
							for (int i = 0; i < Data.length / 6; i++) {
								byte[] array = Arrays.copyOfRange(Data, 6 * i, 6 * i + 6);
								byte[] checkArray = new byte[5];
								System.arraycopy(array, 0, checkArray, 0, 5);
								byte[] ByteCode = new byte[3];
								System.arraycopy(array, 1, ByteCode, 0, 3);
								if (array[5] == ByteUtils.checkByte(checkArray)) {
									DataCode dataCode = new DataCode();
									dataCode.setByteCode(ByteUtils.bytesToHexString(ByteCode));
									listCode.add(dataCode);
								}
							}
							LoraInfo parseData = new LoraInfo();
							parseData.setCityID(ByteUtils.bytesToHexString(cityID));
							parseData.setSendTime(DateUtil.getNowDateTime());
							parseData.setDataList(listCode);
							LoraService dataService = (LoraService) SpringContextUtil.getBean("LoraService");
							dataService.save(parseData);
							//lora数据解析后通过socketClient推送到后端服务（断开需自动重连）
							//backClient.sendMessage(JSON.toJSONString(parseData));
							// out.writeUTF(JSON.toJSONString(parseData));
							//  out.flush();
						}
					}
//					byte[] response = null;
//					logger.info(" 接收数据response============= " + response);
//
//					if (response != null) {
//						AtomicBoolean isWriting = isWritingMap.get(client);
//						if (isWriting == null) {
//							isWriting = new AtomicBoolean(false);
//							isWritingMap.put(client, isWriting);
//						}
//						while (isWriting.get()) {
//							Thread.sleep(5);
//						}
//						isWriting.set(true);
//						client.register(selector, SelectionKey.OP_WRITE, response);
//					} else {
//						logger.info("-----------服务端没有可应答的数据！----------");
//					}
				}
			} catch (Exception e) {
				try {
					SelectionKey selectionKey = client.keyFor(selector);
					if (selectionKey != null) {
						selectionKey.cancel();
					}
					logger.warn("客户端[" + client + "]关闭了连接,原因[" + e + "]");
					if (client != null) {
						client.close();
					}
				} catch (IOException e1) {
					logger.error("客户端[" + client + "]关闭异常" + e1.getMessage());
				}
			} finally {
				AtomicBoolean isReading = isReadingMap.get(client);
				isReading.set(false);
			}
		}

		/**
		 * 对比数据CRC结果
		 * 
		 * @param crcCheckData
		 *            待校验的数据
		 * @param data
		 *            原来去包头（不包含前四位）的数据包。
		 * @return 一致则true,否则false
		 */
		private boolean checkCRCVal(final ByteBuffer crcCheckData, final ByteBuffer data) {

			return false;
		}

		/**
		 * 获取设备在址
		 * 
		 * @param data
		 *            去掉包头（不包含前四位）的数据包
		 * @return 返回设备地址四位byte
		 */
		private byte[] getDevice(ByteBuffer data) {

			return null;
		}

		/**
		 * 获取一个数据指定长度的内容段
		 * 
		 * @param data
		 *            原数据
		 * @param posIndex
		 *            起如位置
		 * @param len
		 *            截取的长度
		 * @return 定长度的内容段 byte[]
		 */
		private byte[] getData(ByteBuffer data, int posIndex, int len) {

			return null;
		}

		/**
		 * 构建应答设备登录（0x41）的数据包
		 * 
		 * @param data
		 *            去掉包头（不包含前四位）的数据包
		 * @return 完整的数据
		 */
		private byte[] buildRegisterPackage(ByteBuffer data) {

			return null;
		}

		/**
		 * 构建应答数据包。
		 * 
		 * @param direction
		 *            标识位，即功能代码
		 * @param device
		 *            设备地址字节数组
		 * @param content
		 *            要应答数据内容
		 * @return
		 */
		private byte[] buildResponsePackage(byte direction, byte[] device, byte[] content) {

			return null;
		}
	}

	public static void main(String[] args) {
		logger.info("--------------->Enter to main method");
		SocketServer server = new SocketServer();
		server.start();
		logger.info("--------------->Server already started!");
	}

}
